package com.dstation.jobhandle;


import cn.hutool.core.text.csv.CsvUtil;
import cn.hutool.core.text.csv.CsvWriter;
import cn.hutool.core.util.CharsetUtil;
import com.alibaba.alink.operator.batch.BatchOperator;
import com.alibaba.alink.operator.batch.recommendation.AlsTrainBatchOp;
import com.alibaba.alink.operator.batch.source.CsvSourceBatchOp;
import com.alibaba.alink.pipeline.recommendation.AlsItemsPerUserRecommender;
import com.alibaba.alink.pipeline.similarity.StringApproxNearestNeighbor;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.dstation.domain.RecommendInfo;
import com.dstation.domain.VideoInfo;
import com.dstation.service.interf.IRecommendService;
import com.dstation.utils.RecommendStrategyUtil;
import com.dstation.utils.RedisPreKey;
import com.xxl.job.core.biz.model.ReturnT;
import com.xxl.job.core.handler.annotation.XxlJob;
import com.xxl.job.core.log.XxlJobLogger;
import lombok.extern.slf4j.Slf4j;

import org.apache.commons.io.FileUtils;
import org.apache.flink.types.Row;
import org.springframework.stereotype.Component;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;
import redis.clients.jedis.Pipeline;

import javax.annotation.Resource;
import java.io.File;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.function.Consumer;
import java.util.regex.Matcher;
import java.util.regex.Pattern;


@Component
@Slf4j
public class SimilarRecommendJob {
    static String REGEX_SPACE = "\r|\n";//定义空格回车换行符
    static Pattern P_SPACE = Pattern.compile(REGEX_SPACE, Pattern.CASE_INSENSITIVE);

    static final String FIELD_DELIMITER = "_!_";
    @Resource
    private IRecommendService recommendService;


    @Resource
    private JedisPool redisPoolFactory;


    @XxlJob("similarRecommendJob")
    public ReturnT<String> similarRecommendJob(String param) throws Exception {
        XxlJobLogger.log("开始---->推荐  实时数据部分 {}", param);

        log.info("开始---->推荐  实时数据部分");
        removeAllRecentRecommendData();

        int dataSize= RecommendStrategyUtil.RECOMMEND_DATA_SIZE;
        try {

            // 构造数据
            final File itemCsvFile = File.createTempFile("item", ".csv");//创建临时文件

            List<VideoInfo> results = recommendService.findAllVideo();
            log.info("---->处理所有的视频，总共 {} 个", results.size());

            List<String> lines = new ArrayList<>();
            for (VideoInfo vo : results) {
                lines.add(vo.getVideoId() + FIELD_DELIMITER + trimSpaceTag(vo.getVideoTitle()) );
            }

            FileUtils.writeLines(itemCsvFile, lines, false);
            log.info("---->写入临时文件 {}", itemCsvFile.getAbsolutePath());


            // 加载CSV,跑算法
            String schema = "videoId int, videoTitle string";
            CsvSourceBatchOp csvSource = new CsvSourceBatchOp().setIgnoreFirstLine(true).setFilePath(itemCsvFile.getAbsolutePath())
                    .setSchemaStr(schema).setFieldDelimiter(FIELD_DELIMITER);

            StringApproxNearestNeighbor pipeline = new StringApproxNearestNeighbor().setIdCol("videoId")
                    .setSelectedCol("videoTitle").setMetric("SIMHASH_HAMMING_SIM").setTopN(60);
            BatchOperator<?> transResult = pipeline.fit(csvSource).transform(csvSource);


            Jedis jedis = redisPoolFactory.getResource();
            Pipeline pipline = jedis.pipelined();

            transResult.distinct().lazyCollect(new Consumer<List<Row>>() {
                @Override
                public void accept(List<Row> d) {
                    log.info("一共对 {} 个商品相似度处理！", d.size());
                    for (Row row : d) {
                        StringBuilder keyName = new StringBuilder(RedisPreKey.SIMILAR_REC_DATA);
                        keyName.append(row.getFieldAs(0).toString());

                        pipline.del(keyName.toString());


                        Map<String, Double> set = new HashMap<String, Double>(31);


                        JSONObject obj = JSON.parseObject(row.getFieldAs(1));
                        JSONArray ars = obj.getJSONArray("ID");
                        JSONArray scs = obj.getJSONArray("METRIC");
                        int source = 1;
                        for (int i = 0; i < ars.size(); i++) {

                            if (row.getFieldAs(0).toString().equals(ars.getString(i))) {
                                continue;
                            }

                            set.put(ars.getString(i), (double) source++);

                        }

//    					boundZSetOperations.add(set);
                        pipline.zadd(keyName.toString(), set);

                    }
                }
            });
            BatchOperator.execute();

            pipline.sync();



            log.info("---->删除临时文件 {}", itemCsvFile.getAbsolutePath());
            itemCsvFile.delete();


        } catch (Exception e) {
            log.error("跑物品相识性出错", e.getMessage());
            XxlJobLogger.log(e);
        }

        XxlJobLogger.log("结束---->推荐  实时数据部分 {}", param);
        log.info("结束 ---->推荐  实时数据部分");


        return ReturnT.SUCCESS;
    }

    private void removeAllRecentRecommendData() {
        boolean isSuccess=recommendService.dropAllRecommendData();

        XxlJobLogger.log("推荐     初始化：｛｝",isSuccess);
        log.info("推荐      初始化：｛｝",isSuccess);
    }




    /**
     * @param str
     * @return 过滤空格
     */
    public static String trimSpaceTag(String str) {
        Matcher m_space = P_SPACE.matcher(str);
        str = m_space.replaceAll(""); // 过滤空格回车标签
        return str.trim(); // 返回文本字符串
    }
}
